package com.pfzy.utildemo.activity.mqtt;

import android.app.AlarmManager;
import android.app.PendingIntent;
import android.app.Service;
import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.os.Handler;
import android.os.IBinder;
import android.util.Log;

import androidx.annotation.Nullable;

import com.pfzy.lib.config.log.MyLog;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

import java.util.Locale;

public class MQService extends Service implements MqttCallback {
    public static final String DEBUG_TAG = "MQService"; // Debug TAG
    public static final String ACTION_START = DEBUG_TAG + ".START"; // 开始
    public static final String ACTION_STOP = DEBUG_TAG + ".STOP"; // 停止
    public static final String ACTION_KEEPALIVE = DEBUG_TAG + ".KEEPALIVE"; // 报警管理器使用的保活
    public static final String ACTION_RECONNECT = DEBUG_TAG + ".RECONNECT"; // 重连

    private Handler mConnHandler;     // 为网络分离处理程序线程
    private MqttDefaultFilePersistence mDataStore; // 默认文件存储
    private MemoryPersistence mMemStore; // 失败时恢复到内存存储
    private MqttConnectOptions mOpts; // 连接选项
    private AlarmManager mAlarmManager; // 用于执行重复任务的报警管理器
    private ConnectivityManager mConnectivityManager; // 检查连接更改
    private static final boolean MQTT_CLEAN_SESSION = true;
    private boolean mStarted = false; // 客户端是否已启动?

    private static final int MQTT_PORT = 1883; // Broker Port
    private static final String MQTT_BROKER = "192.168.80.238"; // Broker URL or IP Address
    private static final String MQTT_URL_FORMAT = "tcp://%s:%d"; // URL Format normally don't change
    private static final java.lang.String CLINET_ID = "liujun";// client id
    private static final String TOPIC = "android06";// a topic
    private static final int MQTT_KEEP_ALIVE = 1000; // KeepAlive Interval in MS

    private MQClient mClient; // Mqtt Client

    /**
     * 保持长连接的主题
     */
    private MqttTopic mKeepAliveTopic;
    private static final String MQTT_KEEP_ALIVE_MESSAGE = "hello word"; // Keep Alive message to send
    public static final int MQTT_QOS_0 = 0; // QOS Level 0 ( 一次无确认 )
    public static final int MQTT_QOS_1 = 1; // QOS Level 1 ( 至少一次并确认 )
    public static final int MQTT_QOS_2 = 2; // QOS Level 2 ( 仅一次，握手确认 )
    private static final int MQTT_KEEP_ALIVE_QOS = MQTT_QOS_0; // Default Keepalive QOS

    /**
     * 服务初始化回调函数
     */
    @Override
    public void onCreate() {
        super.onCreate();

        /**创建一个Handler*/
        mConnHandler = new Handler();
        try {
            /**新建一个本地临时存储数据的目录，该目录存储将要发送到服务器的数据，直到数据被发送到服务器*/
            mDataStore = new MqttDefaultFilePersistence(getCacheDir().getAbsolutePath());
        } catch (Exception e) {
            e.printStackTrace();
            /**新建一个内存临时存储数据的目录*/
            mDataStore = null;
            mMemStore = new MemoryPersistence();
        }
        /**连接的参数选项*/
        mOpts = new MqttConnectOptions();
        /**删除以前的Session*/
        mOpts.setCleanSession(MQTT_CLEAN_SESSION);
        // 不要在mOpts上设置保活间隔，我们会通过警报来跟踪它
        /**定时器用来实现心跳*/
        mAlarmManager = (AlarmManager) getSystemService(ALARM_SERVICE);
        /**管理网络连接*/
        mConnectivityManager = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);
    }

    /**
     * 服务器启动的回调的函数
     *
     * @param intent
     * @param flags
     * @param startId
     * @return
     */
    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
        super.onStartCommand(intent, flags, startId);
        String action = intent.getAction();
        Log.i(DEBUG_TAG, "收到 action of " + action);
        if (action == null) {
            Log.i(DEBUG_TAG, "启动服务而不采取任何行动\n 可能是由于崩溃");
        } else {
            if (action.equals(ACTION_START)) {
                start();/**开始连接*/
            } else if (action.equals(ACTION_STOP)) {
                stop();/**停止连接*/
            } else if (action.equals(ACTION_KEEPALIVE)) {
                keepAlive();/**保持连接*/
            } else if (action.equals(ACTION_RECONNECT)) {
                if (isNetworkAvailable()) {
                    reconnectIfNecessary();/**重新连接*/
                }
            }
        }
        return START_REDELIVER_INTENT;
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    /**
     * 开始连接
     */
    private synchronized void start() {
        /**判断是否已经连接到服务器*/
        if (mStarted) {
            _log("尝试在已启动的情况下启动");
            return;
        }

        /**判断是否还在保持长连接*/
        if (hasScheduledKeepAlives()) {
            /**断开长连接*/
//            stopKeepAlives();
        }

        connect();

        /**注册一个监听网路连接改变的方法*/
        registerReceiver(mConnectivityReceiver,new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));
    }

    /**
     * 尝试停止Mqtt客户端以及停止警报管理器中排队的所有保持活动消息
     */
    private synchronized void stop() {
        if (!mStarted) {
            Log.i(DEBUG_TAG, "尝试停止未运行的连接");
            return;
        }
        if (mClient != null) {
            mConnHandler.post(new Runnable() {
                @Override
                public void run() {
                    try {
                        mClient.disconnect(3000, new DisConnectCallBackHandler());
                    } catch (MqttException ex) {
                        ex.printStackTrace();
                    }
                    mClient = null;
                    mStarted = false;
                    /**停止保持心跳，长连接*/
                    stopKeepAlives();
                }
            });
        }
         unregisterReceiver(mConnectivityReceiver);
    }

    /**
     * 保持长连接
     */
    private synchronized void keepAlive() {
        /**如果还在与ActiveMQ服务器连接*/
        if (isConnected()) {
            try {
//                startKeepAlives();
                /**发送一个心跳包，保持与服务器长连接*/
                sendKeepAlive();
                return;
            } catch (MqttConnectivityException ex) {
                ex.printStackTrace();
                /**如果保持长连接出现异常，重新连接*/
                reconnectIfNecessary();
            } catch (MqttPersistenceException ex) {
                ex.printStackTrace();
                stop();
            } catch (MqttException ex) {
                ex.printStackTrace();
                stop();
            }
        }
    }

    /**
     * 检查网络是否可用
     */
    private boolean isNetworkAvailable() {
        NetworkInfo info = mConnectivityManager.getActiveNetworkInfo();
        return (info == null) ? false : info.isConnected();
    }

    /**
     * 检查连接情况，如果没有连接就自动连接
     */
    private synchronized void reconnectIfNecessary() {
        if (mStarted && mClient == null) {
            connect();
        }
    }

    /**
     * 查询报警管理器以检查当前是否安排了保活
     */
    private synchronized boolean hasScheduledKeepAlives() {
        Intent i = new Intent();
        i.setClass(this, MQService.class);
        i.setAction(ACTION_KEEPALIVE);
        PendingIntent pi = PendingIntent.getBroadcast(this, 0, i, PendingIntent.FLAG_NO_CREATE);
        return (pi != null) ? true : false;
    }

    /**
     * 连接到具有适当数据存储的broker
     */
    private synchronized void connect() {
        /**创建一个url*/
        String url = String.format(Locale.US, MQTT_URL_FORMAT, MQTT_BROKER, MQTT_PORT);
        _log("Connecting with URL: " + url);

        /**创建一个MQClient*/
        try {
            if (mDataStore != null) {
                _log("连接 with DataStore");
                mClient = new MQClient(url, CLINET_ID, mDataStore);
            } else {
                _log("连接 with MemStore");
                mClient = new MQClient(url, CLINET_ID, mMemStore);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }

        mConnHandler.post(new Runnable() {
            @Override
            public void run() {
                try {
                    mClient.connect(mOpts, new ConnectCallBackHandler());
                    /**开始订阅*/
                    mClient.subscribe(TOPIC, 0);
                    /**订阅的回调*/
                    mClient.setCallback(MQService.this);
                    mStarted = true; // 服务现已连接
                    Log.i(DEBUG_TAG, "已成功连接并订阅开始保持活动");
                    /**保持长连接*/
                    startKeepAlives();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 开始定时
     * 计划通过报警管理器中的挂起意向保持有效
     */
    private void startKeepAlives() {
        Intent i = new Intent();
        i.setClass(this, MQService.class);
        i.setAction(ACTION_KEEPALIVE);
        PendingIntent pi = PendingIntent.getService(this, 0, i, 0);
        mAlarmManager.setRepeating(AlarmManager.RTC_WAKEUP, System.currentTimeMillis() + MQTT_KEEP_ALIVE, MQTT_KEEP_ALIVE, pi);
    }

    /**
     * 取消定时
     * 取消报警管理器中的挂起意图
     */
    private void stopKeepAlives() {
        Intent i = new Intent();
        i.setClass(this, MQService.class);
        i.setAction(ACTION_KEEPALIVE);
        PendingIntent pi = PendingIntent.getService(this, 0, i, 0);
        mAlarmManager.cancel(pi);
    }

    /**
     * 检查与AcitiveMQ服务器的网络连接状态
     * @return 如果匹配，则为true如果未连接，则为false
     */
    private boolean isConnected() {
        if (mStarted && mClient != null && !mClient.isConnected()) {
            Log.i(DEBUG_TAG, "我们认为有联系的东西和有联系的之间不匹配");
        }
        if (mClient != null) {
            return (mStarted && mClient.isConnected()) ? true : false;
        }
        return false;
    }

    /**
     * 发送一个心跳包，保持长连接
     * @return MqttDeliveryToken 指定的令牌，您可以选择等待完成
     */
    private synchronized MqttDeliveryToken sendKeepAlive()
            throws MqttConnectivityException, MqttPersistenceException, MqttException {
        if (!isConnected())
            throw new MqttConnectivityException();

        if (mKeepAliveTopic == null) {
            mKeepAliveTopic = mClient.getTopic(TOPIC);
        }
        Log.i(DEBUG_TAG, "Sending Keepalive to " + MQTT_BROKER);

        MqttMessage message = new MqttMessage(MQTT_KEEP_ALIVE_MESSAGE.getBytes());
        message.setQos(MQTT_KEEP_ALIVE_QOS);
        /**发送一个心跳包给服务器，然后回调到：messageArrived 方法中*/
        return mKeepAliveTopic.publish(message);
    }

    /**
     * 通过ConnectivityManager监听连接通道的接收器
     */
    private final BroadcastReceiver mConnectivityReceiver = new BroadcastReceiver() {
        @Override
        public void onReceive(Context context, Intent intent) {
            Log.i(DEBUG_TAG, "Connectivity Changed...");
        }
    };

    /**
     * 连接失败
     */
    @Override
    public void connectionLost(Throwable cause) {
        stopKeepAlives();
        mClient = null;
        /**检查网络是否可用*/
        if (isNetworkAvailable()) {
            reconnectIfNecessary();
        }
    }

    /**
     * 接收到服务器推送来的消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        Log.d(DEBUG_TAG, "============messageArrived==========");
        Log.d(DEBUG_TAG, topic + " = " + new String(message.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        Log.d(DEBUG_TAG, "============deliveryComplete==========");
    }

    private class MqttConnectivityException extends Exception {
    }

    /**
     * 开始连接ActiveMQ服务器
     */
    public static void actionStart(Context ctx) {
        Intent i = new Intent(ctx, MQService.class);
        i.setAction(ACTION_START);
        ctx.startService(i);
    }

    /**
     * 停止连接ActiveMQ服务器
     */
    public static void actionStop(Context ctx) {
        Intent i = new Intent(ctx, MQService.class);
        i.setAction(ACTION_STOP);
        ctx.startService(i);
    }

    private void _log(String msg) {
        MyLog.logd("MQService: " + msg);
    }
}
